package day07


import java.sql.Struct

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._

/**
  * UDAF即UserDefinedAggregateFunction，是用户自定义进行聚合操作的。
  */
class PersonWCUDAF extends UserDefinedAggregateFunction{
  // inputSchema定义输入类型
  override def inputSchema: StructType = {
    //固定格式,注意要在StructField("count", StringType, true)填对应的(字段名，类型，是否可以为空)
    StructType(Array(StructField("str", StringType, true)))
  }
  // bufferSchema定义缓冲类型
  override def bufferSchema: StructType = {
    //后面的buffer就是在缓冲区，代表后面的(buffer的名，类型，是否可以为空)
    StructType(Array(StructField("count", IntegerType, true)))
  }
  // dataType定义输出类型（返回值类型）
  override def dataType: DataType = IntegerType   //要返回相同key对应的数量，所以返回IntegerType
  // deterministic,是否是确定的？默认为false。如果为true就代表返回值和输入值是相同的，如果为false就代表不同。
  override def deterministic: Boolean = true
  // initialize初始化操作方法，可以在这个方法中进行初始化
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    //定义下标为0的数据初始值为0，后期的累加都把累加到下标为0的这里面,
    buffer(0) = 0 //即相同的key对应的数值为0
  }
  // update局部聚合（分区内的聚合）,input指传进来的新值，注意是新的值
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    //后期的累加都把累加到下标为0的buffer里面。 buffer.getAs(0)为获取下标为0的值，需要给个泛型即buffer.getAs[Int](0)
    buffer(0) = buffer.getAs[Int](0) + 1  //这里intput :Row代表新进来的值，因为我要做的是来一个数据就+1，所以这里没用到input
  }
  // merge全局聚合。buffer1、buffer2代表各分区的聚合结果
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)    //各个分区里面取值累加然后在赋给某个buffer
  }
  // evaluate：在这个buffer中可以进行其他的操作，比如给他加某个值，或者给他截取等操作
  override def evaluate(buffer: Row): Any = buffer.getAs[Int](0)    //还可以给后面加个 | 这样的符号，之后就可以通过|截取数据。
}
